

import com.typesafe.scalalogging.Logger
import entity.LogBean
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import scala.collection._
import scala.util.parsing.json
import scala.util.parsing.json.JSON
/**
  *
  * @author Abyss
  * @date 2019/8/30
  * @description
  */
object ReadKafka {

//
//  private val logger = Logger(this.getClass)
//
//  def main(args: Array[String]): Unit = {
//    System.setProperty("HADOOP_USER_NAME", "hadoop")
//
//    val conf = new SparkConf().setAppName("ReadKafka").setMaster("local[*]")
//
//    val spark = SparkSession.builder().config(conf).getOrCreate()
//    val streamContext = new StreamingContext(spark.sparkContext, Seconds(5))
//    val df = spark
//      .readStream
//      .format("kafka")
//      .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
//      .option("subscribe", "topic1")
//      .load()
//    //直连方式相当于跟kafka的Topic至直接连接
//    //"auto.offset.reset:earliest(每次重启重新开始消费)，latest(重启时会从最新的offset开始读取)
//    val kafkaParams = immutable.Map[String, Object](
//      "bootstrap.servers" -> "192.168.0.185:9092",
//      "key.deserializer" -> classOf[StringDeserializer],
//      "value.deserializer" -> classOf[StringDeserializer],
//      "group.id" -> "spark",
//      "auto.offset.reset" -> "latest",
//      "enable.auto.commit" -> (false: java.lang.Boolean)
////      "sasl.mechanism"->"PLAIN",
////      "security.protocol"->"SASL_PLAINTEXT"
//    )
//    //配置文件路径
////    System.setProperty("java.security.auth.login.config", "/Users/abyss/Dev/goshine/sparkCS/src/main/resources/kafka_client_jaas.conf")
//    val topics = Array("mq_cj_data_1")
//
//    val kafkaDStream = KafkaUtils.createDirectStream[String, String](
//      streamContext,
//      LocationStrategies.PreferConsistent,
//      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
//    )
//
//    //如果使用SparkStream和Kafka直连方式整合，生成的kafkaDStream必须调用foreachRDD
//    kafkaDStream.foreachRDD(kafkaRDD => {
//      if (!kafkaRDD.isEmpty()) {
//        //获取当前批次的RDD的偏移量
//        val offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
//        //拿出kafka中的数据
//        val lines = kafkaRDD.map(_.value())
//        //将lines字符串转换成json对象
//        val logBeanRDD = lines.map(line => {
//          var logBean: LogBean = null
//          try {
////            logBean=JSON.parse(line).asInstanceOf[LogBean]
//            val parse_str = str_json(line)
//            logBean=LogBean(parse_str("dataPointId").asInstanceOf[Int], parse_str("dataSource").asInstanceOf[Int],
//              parse_str("dataType").asInstanceOf[Int], parse_str("dataTime").asInstanceOf[Long]
//            , parse_str("data").asInstanceOf[immutable.Map[String, String]])
//
//          } catch {
//            case e: Exception => {
//              //logger记录
//              logger.error("json解析错误！line:" + line, e)
//            }
//          }
//          logBean
//        })
//
//        //过滤
//        val filteredRDD = logBeanRDD.filter(_ != null)
//
//        //将RDD转化成DataFrame,因为RDD中装的是case class
//        import spark.implicits._
//
//        val df = filteredRDD.toDF()
//
//        df.show()
//        //将数据写到hdfs中:hdfs://hd1:9000/360
//        //        df.repartition(1).write.mode(SaveMode.Append).parquet(args(0))
//
//        //提交当前批次的偏移量，偏移量最后写入kafka
//        kafkaDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
//      }
//    })
//
//    //启动
//    streamContext.start()
//    streamContext.awaitTermination()
//    streamContext.stop()
//
//  }
//
//
//  def str_json(string_json:String):collection.immutable.Map[String,Any]=
//  {
//    var first :collection.immutable.Map[String, Any] = collection.immutable.Map()
//    val jsonS = JSON.parseFull(string_json)
//    //不确定数据的类型时，此处加异常判断
//    if (jsonS.isInstanceOf[Option[Any]]){
//      first = regJson(jsonS)
//    }
//    first
//  }
//
//
//  def regJson(json:Option[Any]) = json match {
//    //转换类型
//    case Some(map:collection.immutable.Map[String, Any]) => map
//    case _ => {println("匹配失败");null}//匹配失败返回Null值
//  }
}

